# --------------------------------------------------------------------------------------------------------------------
# Project:             Digital Payment System (DPS)
# Website:             http://code.google.com/p/dps-x509/
# Purpose of document: Web service test tool for registrar
# --------------------------------------------------------------------------------------------------------------------
import urllib, sys
sys.path.append("..")

from time import time

try:
  import simplejson as json
except ImportError:
  import json
  
import dps.x509Asn1Creator
import dps.x509AttributeAndValue
import dps.x509Helper
import dps.x509CertificateParams
import dps.utils

import testCertificateData
  
HOST = ""
SPEW = False


# --------------------------------------------------------------------------------------------------------------------
def append(l, s):
  """ append l to s and return l. will also print if SPEW == True """
  l.append(s)
  
  global SPEW
  if SPEW:
    print(s)
  return l

# --------------------------------------------------------------------------------------------------------------------
def callWebService(name, params, exp):
  """ call the web service and compare expected and actual results
      @param name   - name of the web service to call
      @param params - struct of input parameters
      @param exp    - json str formatted expected results
      @return success, message tuple. success = true if expected result = actual result """
  #print ("executing "+name+" ...");
  res = False
  message = ""
  try:
    global HOST
    #print len(params)
    p = urllib.urlencode(params)
    #print params    
    f = urllib.urlopen(HOST + name, p)
    actual = urllib.unquote(f.read())
    f.close()
    if actual == exp:
      res = True
    else:
      message = "failed. actual len: "+str(len(actual))+" expected:"+str(len(exp))+"\n"
      message += "actual: \n"+actual+"\n expected: \n"+exp
  except:
    message = "Error connecting to web service"
  return res, message    
  
# --------------------------------------------------------------------------------------------------------------------
class TestSuite:
  """ container holding all tests """
  def __init__(self):
    self.tests_ = []
    
  def addTest(self, t):
    """ add a test to the suite. 
    @param t Test must be a bool,string returning function that takes no params """
    self.tests_.append(t)
    
  def runTests(self):
    """ run all tests in suite """
    passed = 0    
    o = []
    o = append(o, "********************************")
    total0 = time()
    for t in self.tests_:
      msg = ""      
      didPass = False
      resText = "False"
      t0 = time()
      try:
        didPass, msg = t()
        if didPass:
          passed += 1
          resText = "True"
      except Exception, e:
        msg = "An exception occured: " + str(sys.exc_info()[0]) + " " + str(e)
      t1 = time()        
      o = append(o, "**** %s: %s : %d seconds" % (t.__name__, resText, t1 - t0))
      if not didPass:
        msg = " "*4 + msg.replace("\n", "\n"+" "*4)
        o = append(o, msg)
    total1 = time()    
    o = append(o, "********************************")
    o = append(o, "**** Tests passed: %d of %d : %d seconds" % (passed, len(self.tests_), total1 - total0))
    o = append(o, "********************************")
    return o

# --------------------------------------------------------------------------------------------------------------------
# --------------------------------------------------------------------------------------------------------------------
def clearCertificate():
  result = json.dumps({"returnCode": "True"})  
  isMatch, msg = callWebService("registrar/clear", {"format": "json"}, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def removeCertificate(cert):
  result = json.dumps({"returnCode": "AllRemovedFromRegistrar"})  
  isMatch, msg = callWebService("registrar/remove", {"format": "json", "certificate": cert}, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def findCertificate(cert, isExpectedInRegistrar=True):
  result = json.dumps({"returnCode": "True"})  
  if not isExpectedInRegistrar:
    result = json.dumps({"returnCode": "False"})  
  isMatch, msg = callWebService("registrar/exactFind", {"format": "json", "certificate": cert}, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64_2048_lengthPublicKeyGood():
  removeCertificate(testCertificateData.testbase64_crt())
  
  params = {"certificate":testCertificateData.testbase64_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def exactMatchExists():
  base64_2048_lengthPublicKeyGood()

  isOk, msg = findCertificate(testCertificateData.testbase64_crt(), True)
  return isOk, msg

# --------------------------------------------------------------------------------------------------------------------
def exactMatchNoExists():
  removeCertificate(testCertificateData.testbase64_crt())

  isOk, msg = findCertificate(testCertificateData.testbase64_crt(), False)
  return isOk, msg

# --------------------------------------------------------------------------------------------------------------------
def exactMatchOnCertificateChain():
  result = json.dumps({"returnCode": "Rejected:InvalidParametersError"})
  params = {"format": "json", "certificate": testCertificateData.mobileMinutesWithChain_crt()}
  isOk, msg = callWebService("registrar/exactFind", params, result)
  return isOk, msg

# --------------------------------------------------------------------------------------------------------------------
def removeNoExists():
  isOk, msg = exactMatchNoExists()
  if not isOk:
    return False, "exactMatchNoExists failed"
  
  return removeCertificate(testCertificateData.testbase64_crt())

# --------------------------------------------------------------------------------------------------------------------
def removeExists():
  isOk, msg = exactMatchExists()
  if not isOk:
    return False, "exactMatchExists failed"
  
  return removeCertificate(testCertificateData.testbase64_crt())

# --------------------------------------------------------------------------------------------------------------------
def removeCertificateChain():
  return removeCertificate(testCertificateData.mobileMinutesWithChain_crt())

# --------------------------------------------------------------------------------------------------------------------
def certificateAlreadyInRegistrar():
  
  isOk, msg = findCertificate(testCertificateData.testbase64_crt())
  if not isOk:    
    isOk, msg = base64_2048_lengthPublicKeyGood()
    if not isOk:
      return False, "Could not add first certificate"
    
  params = {"certificate":testCertificateData.testbase64_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:CertificateAlreadyInRegistrarError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64_1024_lengthPublicKeyGood():
  isOk, msg = removeCertificate(testCertificateData.testbase64_1024pubkey_crt())
  if not isOk:
    return False, "Unable to remove certificate"
  
  params = {"certificate":testCertificateData.testbase64_1024pubkey_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64dsa():
  params = {"certificate":testCertificateData.testbase64dsa_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:PKIMethodNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64sha256():
  params = {"certificate":testCertificateData.testbase64sha256_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:SignatureAlgorithmNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64sha384():
  params = {"certificate":testCertificateData.testbase64sha384_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:SignatureAlgorithmNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64sha512():
  params = {"certificate":testCertificateData.testbase64sha512_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:SignatureAlgorithmNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64md2():
  params = {"certificate":testCertificateData.testbase64md2_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:SignatureAlgorithmNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64md5():
  params = {"certificate":testCertificateData.testbase64md5_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:SignatureAlgorithmNotSupportedError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64BadSignature():
  params = {"certificate":testCertificateData.testbase64switchchars_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:InvalidSignatureError"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def base64ExtraCharacters():
  params = {"certificate":testCertificateData.testbase64extra_crt(), "format": "json"}
  result = json.dumps({"returnCode": "Rejected:DecodingError"})
  
  certChain = dps.x509Helper.certsToBase64(testCertificateData.testbase64extra_crt())
  dps.utils.assertion(len(certChain) == 1, "1 certificate expected")
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def derGood():
  cert = open("../certificates/testder.crt", "rb").read()  

  certChain = dps.x509Helper.certsToBase64(cert)

  isOk, msg = removeCertificate(certChain[0])
  if not isOk:
    return False, "Could not clear registry"

  params = {"certificate":certChain[0], "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isOk, msg = callWebService("registrar/upload", params, result);
  return isOk, msg
  
# --------------------------------------------------------------------------------------------------------------------
def chain2UploadedAtSameTime():
  cert = testCertificateData.mobileMinutesWithChain_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 2, "Must be two certificates in the chain")

  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"  
  
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain2UploadedIndividually():
  """feed each cert into the registrar individually"""
  cert = testCertificateData.mobileMinutesWithChain_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 2, "Must be two certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"  
  
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch = False
  msg = ""
  certChain.reverse()
  for c in certChain:
    isMatch, msg = callWebService("registrar/upload", {"certificate":c, "format": "json"}, result)
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedAtSameTime():
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")

  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"  
  
  params = {"certificate":cert, "format": "json"}
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch, msg = callWebService("registrar/upload", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedIndividually():
  """feed each cert into the registrar individually"""
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"
  
  result = json.dumps({"returnCode": "Valid"})
  
  isMatch = False
  msg = ""
  certChain.reverse()
  for c in certChain:
    isMatch, msg = callWebService("registrar/upload", {"certificate":c, "format": "json"}, result)
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedStaggered1():
  """"""
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"
  
  result = json.dumps({"returnCode": "Valid"})
  isMatch, msg = callWebService("registrar/upload", {"certificate":certChain[2], "format": "json"}, result)
  if not isMatch:
    return False, "Unable to load first cert"
  
  isMatch, msg = chain3UploadedAtSameTime()
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedStaggered2():
  """"""
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"
  
  result = json.dumps({"returnCode": "Valid"})
  isMatch, msg = callWebService("registrar/upload", {"certificate":certChain[2], "format": "json"}, result)
  if not isMatch:
    return False, "Unable to load first cert"

  isMatch, msg = callWebService("registrar/upload", {"certificate":certChain[1], "format": "json"}, result)
  if not isMatch:
    return False, "Unable to load second cert"
  
  isMatch, msg = chain3UploadedAtSameTime()
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedStaggered3():
  """"""
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"
  
  result = json.dumps({"returnCode": "Valid"})
  isMatch, msg = callWebService("registrar/upload", {"certificate":certChain[2], "format": "json"}, result)
  if not isMatch:
    return False, "Unable to load first cert"

  last2certs = dps.x509Helper.prettyBase64Certificates(certChain[:-1])
  isMatch, msg = callWebService("registrar/upload", {"certificate":last2certs, "format": "json"}, result)
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def chain3UploadedIssuerCertNotInRegistrar():
  """"""
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")
  
  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"
  
  result = json.dumps({"returnCode": "Rejected:IssuerPublicKeyNotFoundError"})
  isMatch, msg = callWebService("registrar/upload", {"certificate":certChain[0], "format": "json"}, result)
  return isMatch, msg


# --------------------------------------------------------------------------------------------------------------------
def getDoubleSpendCerts():
  cparams = dps.x509CertificateParams.X509CertificateParams()
  cissuer = dps.x509AttributeAndValue.X509AttributeAndValueList()
  cissuerItem = dps.x509AttributeAndValue.X509AttributeAndValue("c")
  cissuer.addItem(cissuerItem)
  cparams.setIssuer(cissuer)
  cparams.setSubject(cissuer)  
  cparams.tokenType_ = "My token"
  cparams.tokenQuantity_ = 14.21
  cparams.registrar_ = "http://myredeemer.com"
  cparams.redeemer_ = "http://registrar.com"
  
  c = dps.x509Asn1Creator.X509Asn1Creator(cparams, testCertificateData.privateKey1())
  c.signWithPrivateKey(c.getRsaPrivateKey())
  
  #create issuer
  aparams = dps.x509CertificateParams.X509CertificateParams()
  aissuer = dps.x509AttributeAndValue.X509AttributeAndValueList()
  aissuerItem = dps.x509AttributeAndValue.X509AttributeAndValue("a")
  aissuer.addItem(aissuerItem)
  aparams.setIssuer(cparams.getSubject())
  aparams.setSubject(aissuer)  
  
  #reassign to next
  a = dps.x509Asn1Creator.X509Asn1Creator(aparams, testCertificateData.privateKey2())
  a.signWithPrivateKey(c.getRsaPrivateKey())

  #create issuer
  bparams = dps.x509CertificateParams.X509CertificateParams()
  bissuer = dps.x509AttributeAndValue.X509AttributeAndValueList()
  bissuerItem = dps.x509AttributeAndValue.X509AttributeAndValue("b")
  bissuer.addItem(bissuerItem)
  bparams.setIssuer(cparams.getSubject())
  bparams.setSubject(bissuer)  
  
  #reassign to next
  b = dps.x509Asn1Creator.X509Asn1Creator(bparams, testCertificateData.privateKey3())
  b.signWithPrivateKey(c.getRsaPrivateKey())
  
  return a.getBase64(), b.getBase64(), c.getBase64()

# --------------------------------------------------------------------------------------------------------------------
def doubleSpendIndividual():
  """ 1. create certificate C and upload it to the registrar
      2. reassign C certificate to A and upload it to the registrar
      3. reassign C certificate to B and upload it to the registrar
  """
  
  a, b, c = getDoubleSpendCerts()
  
  #c
  isMatch, msg = removeCertificate(c)
  if not isMatch:
    return False, "Unable to remove certificate"

  isMatch, msg = callWebService("registrar/upload", 
                                {"certificate":c, "format": "json"}, 
                                json.dumps({"returnCode": "Valid"}))
  if not isMatch:
    return False, "Could not create first certificate"
  
  #a
  isMatch, msg = removeCertificate(a)
  if not isMatch:
    return False, "Unable to remove certificate"  
  
  isMatch, msg = callWebService("registrar/upload", 
                                {"certificate":a, "format": "json"}, 
                                json.dumps({"returnCode": "Valid"}))
  if not isMatch:
    return False, "Could not reassign certificate"

  #b
  isMatch, msg = removeCertificate(b)
  if not isMatch:
    return False, "Unable to remove certificate"  

  isMatch, msg = callWebService("registrar/upload", 
                                {"certificate":b, "format": "json"}, 
                                json.dumps({"returnCode": "Rejected:DoubleSpendError"}))
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def doubleSpendAtSameTime():
  """ 1. create certificate C and upload it to the registrar
      2. reassign C certificate to A and upload it to the registrar
      3. reassign C certificate to B and upload it to the registrar
  """
  
  a, b, c = getDoubleSpendCerts()
  
  isMatch, msg = removeCertificate(c)
  if not isMatch:
    return False, "Unable to remove certificate"
  isMatch, msg = removeCertificate(b)
  if not isMatch:
    return False, "Unable to remove certificate"
  isMatch, msg = removeCertificate(a)
  if not isMatch:
    return False, "Unable to remove certificate"

  firstChain = dps.x509Helper.prettyBase64Certificates([b,c])
  isMatch, msg = callWebService("registrar/upload", 
                                {"certificate":firstChain, "format": "json"}, 
                                json.dumps({"returnCode": "Valid"}))
  if not isMatch:
    return False, "Could not create first certificate"
  
  secondChain = dps.x509Helper.prettyBase64Certificates([a,c])
  isMatch, msg = callWebService("registrar/upload", 
                                {"certificate":secondChain, "format": "json"}, 
                                json.dumps({"returnCode": "Rejected:DoubleSpendError"}))
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def searchMatch():
  isOk, msg = chain3UploadedAtSameTime()
  if not isOk:
    return False, "Error in chain3UploadedAtSameTime"  
  
  certChain = dps.x509Helper.prettyBase64Certificates(dps.x509Helper.certsToBase64(testCertificateData.certChain3_crt()))
  
  params = {"tokenType": "Dealer tokens", "format": "json"}
  exp = {"returnCode": "Valid", 
         "certificateChains":
         [{"tokenType": "Dealer tokens",
           "redeemer": "http://someRedeemer.com", 
           "registrar": "http://localhost:8083/registrar", 
           "tokenQuantity": "5.00", 
           "exchanger": "http://localhost:8083/registrar",
           "certificateChain": certChain}]}

  result = json.dumps(exp)
  
  isMatch, msg = callWebService("registrar/search", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
def searchNoMatch():
  cert = testCertificateData.certChain3_crt()
  certChain = dps.x509Helper.certsToBase64(cert)
  dps.utils.assertion(len(certChain) == 3, "Must be three certificates in the chain")

  isOk, msg = removeCertificate(cert)
  if not isOk:
    return False, "Unable to remove certificate"  
  
  params = {"issuer":"d", "format": "json"}
  result = json.dumps({"returnCode": "Valid", "certificateChains":[]})
  
  isMatch, msg = callWebService("registrar/search", params, result);
  return isMatch, msg

# --------------------------------------------------------------------------------------------------------------------
# --------------------------------------------------------------------------------------------------------------------
def runTests(host = "http://dps-x509.appspot.com/", spewOutput = False):
  """ execute the tests in the test suite. the default params are configured for the web gui interfaces
  @param host - where to run the app
  @param spewOutput - print output as the tests are running. This is useful when running locally, but isn't wanted when running via the gui web interface. 
  @return a list of string results. """
  global HOST
  global SPEW
  
  HOST = host
  SPEW = spewOutput
    
  out = []
  if SPEW:
    append(out, "host: "+HOST)  
  ts = TestSuite()

  #exact match
  ts.addTest(exactMatchExists)
  ts.addTest(exactMatchNoExists)
  ts.addTest(exactMatchOnCertificateChain)

  #remove tests
  ts.addTest(removeExists)
  ts.addTest(removeNoExists)
  ts.addTest(removeCertificateChain)
  
  #single upload tests
  ts.addTest(base64_2048_lengthPublicKeyGood)
  ts.addTest(base64_1024_lengthPublicKeyGood)
  ts.addTest(certificateAlreadyInRegistrar)
  ts.addTest(base64dsa)
  ts.addTest(base64sha256)
  ts.addTest(base64sha384)
  ts.addTest(base64sha512)
  ts.addTest(base64md2)
  ts.addTest(base64md5)
  ts.addTest(base64BadSignature)
  ts.addTest(base64ExtraCharacters)
  ts.addTest(derGood) 
  
  #chain tests
  ts.addTest(chain2UploadedAtSameTime)
  ts.addTest(chain2UploadedIndividually)
  ts.addTest(chain3UploadedAtSameTime)
  ts.addTest(chain3UploadedIndividually)
  ts.addTest(chain3UploadedStaggered1)
  ts.addTest(chain3UploadedStaggered2)
  ts.addTest(chain3UploadedStaggered3)
  ts.addTest(chain3UploadedIssuerCertNotInRegistrar)
  
  #double spend
  ts.addTest(doubleSpendIndividual)
  ts.addTest(doubleSpendAtSameTime)
  
  #search
  ts.addTest(searchMatch)
  ts.addTest(searchNoMatch)
  
  out += ts.runTests()
  
  return out
  
# --------------------------------------------------------------------------------------------------------------------
if __name__ == "__main__":
  runTests("http://dps-x509.appspot.com/", True)
  #runTests("http://localhost:8083/", True)

